#include "Dispatch.hpp"
#include "Reactor.hpp"
#include "Socket.hpp"
#include "ConnPool.hpp"
#include "../base/BaseLog.hpp"

#include <cstring>
#include <sys/select.h>
#include <ctime>
#include <sys/types.h>
#include <unistd.h>
#include <memory>

using namespace DBServer::Socket;


// monitor the listen fd by blocking,
// when there is a client connect,
// Select() return
static int Select(int listen_fd) {
    fd_set rset;
    int retval= 0;

    FD_ZERO(&rset);
    FD_SET(listen_fd, &rset);

    // block to wait client
    retval = select(10, &rset, nullptr, nullptr, nullptr);

    if (retval == -1)
    {
        if (errno == EINTR) 
        {
            warn("a singal was caught.\n");
            return 1;
        }
        else 
        {
            error("select fail!\n");
            return -1;
        }
    }
    return 0;
}


/*---------------------------------------------
 * 
 * Dispatch Api Immpeletement
 * 
 * --------------------------------------------
*/

Dispatch::Dispatch(Listener& le) 
    : status(FREE), rig_reactors(0), listener(&le), is_stop(false)
{
    memset(reactors, 0, sizeof(reactors));
}

Dispatch::~Dispatch() {
    debug("");
}

int
Dispatch::run(std::function<std::string(std::string&&)> _logic_func) {
    if (listener == nullptr)
    {
        error("what(): DBServer::Socket::Dispatch::listener is a null pointer!");
        return -1;
    }

    if (listener->get_socket_status() != LISTENING)
    {
        error("what():the socekt is not listening\n");
        return -1;
    }

    if (rig_reactors == 0)
    {
        error("what():there is no reactor registe in this dispatch!\n");
        return -1;
    }
    
    if (status == RUNING)
    {   
        error("what():this dispatch is already running\n");
        return -1;
    }

    // log
    info("dbserver running!\n");
    
    int basic = -1;
    int res = 0;
    while (!is_stop)
    {
        if ((res = Select(listener->get_socket_fd())) != 0) 
        {
            if (res == 1) continue;
            return -1;
        }
        // get a ServerConn from the object pool and initalize it
        std::shared_ptr<ServerConn> conn;
        conn = ConnPool::GetInstance()->get();
        if (conn == nullptr)
        {
            // there is no Connection in the object pool
            continue;
        }
        // set socket attribution
        conn->create_socket(listener->get_socket_fd());
        conn->set_logic_cb(_logic_func);
        socket_set_unblock(*conn);
        // dispatch
        dispatch_conn(conn, basic);
    }
    return 0;
}

int
Dispatch::stop() {
    is_stop = true;
    info("Dispatch::run() ending!\n");
    return 0;
}

int
Dispatch::dispatch_conn(std::shared_ptr<ServerConn> conn, int& basic) {
    // load balancing in the reactors 
    int location = ++basic % rig_reactors;
    // register event into the reactors
    reactors[location]->register_read(conn.get());
    reactors[location]->register_rdhup(conn.get());
    reactors[location]->add_conn(conn);
    return 0;
}

int
Dispatch::reactor_reg(Reactor* reactor) {
    info("reactor(%d)\n", rig_reactors);

    reactors[rig_reactors] = reactor;
    ++rig_reactors;
    return 0;
}

int
Dispatch::register_read(Listener* listener) {
    Reactor::register_read(listener);
    return 0;
}

int
Dispatch::delete_event(Listener* listener) {
    Reactor::delete_event(listener);
    return 0;
}

int
Dispatch::get_rig_reactos() {
    return rig_reactors;
}

int 
Dispatch::get_listener_fd() {
    return listener->get_socket_fd();
}